package com.seaboxdata.jt809.kafka;

import com.alibaba.fastjson.JSONObject;
import com.seaboxdata.jt809.common.util.PropertiesUtil;
import com.seaboxdata.jt809.packet.JT809Packet0x1202;
import java.util.Date;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * @ClassName MessageProducer
 * @Description 消息生产者类，用于将消息同步到kafka
 * @Author DFJX
 * @Company 北京东方金信科技有限公司
 * @Date 2020/7/20 17:34
 *         Version 1.0
 **/
public class JT809Packet0x1202MessageProducer {

    private Producer<String, String> producer;

    private static boolean isWin = System.getProperty("os.name").toLowerCase().contains("win");

    private String kafkaServer = PropertiesUtil.getProperty("kafka.server");

    private String authConfig = PropertiesUtil.getProperty("kafka.jaas.config");

    private String authWinConfig = PropertiesUtil.getProperty("kafka.win.jaas.config");

    private String krb5Config = PropertiesUtil.getProperty("krb5.config");

    private String krb5WinConfig = PropertiesUtil.getProperty("krb5.win.config");

    private String isKerberes = PropertiesUtil.getProperty("kafka.kerberos.enable");


    private String topic = PropertiesUtil.getProperty("kafka.topic");

    public JT809Packet0x1202MessageProducer() {
        Properties props = new Properties();
        if ("true".equals(isKerberes)) {
            if (isWin) {
                //如果是windows指定本地的配置文件,本地调试使用
                this.authConfig = authWinConfig;
                this.krb5Config = krb5WinConfig;
            }
            System.setProperty("java.security.auth.login.config", authConfig);
            System.setProperty("java.security.krb5.conf", krb5Config);
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "kafka");
            props.put("sasl.mechanism", "GSSAPI");
        }
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1000);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producer = new KafkaProducer<String, String>(props);
    }

    public void sendMessageToKafka(JT809Packet0x1202 obj) {
        obj.setCjsj(new Date());
        producer.send(new ProducerRecord<String, String>(topic, JSONObject.toJSONString(obj)));
    }
}
